package etcd

import (
	"context"
	"encoding/json"
	"fmt"
	"go-rpc-2/server_register_and_found/register"
	clientv3 "go.etcd.io/etcd/client/v3"
	"go.etcd.io/etcd/client/v3/concurrency"
	"log"
	"sync"
)

type Register struct {
	client  *clientv3.Client
	sess    *concurrency.Session
	cancels []context.CancelFunc
	mu      sync.Mutex
}

func (r *Register) RegisterService(ctx context.Context, si register.ServiceInstance) error {
	instanceName := getInstanceName(si.ServiceName, si.Addr)

	bs, err := json.Marshal(si)
	if err != nil {
		return err
	}

	_, err = r.client.Put(ctx, instanceName, string(bs), clientv3.WithLease(r.sess.Lease()))
	if err != nil {
		return err
	}
	return nil
}

func (r *Register) UnRegisterService(ctx context.Context, si register.ServiceInstance) error {
	instanceName := getInstanceName(si.ServiceName, si.Addr)
	_, err := r.client.Delete(ctx, instanceName)
	return err
}

func (r *Register) ListService(ctx context.Context, serviceName string) ([]register.ServiceInstance, error) {
	serviceName = getServiceName(serviceName)
	resp, err := r.client.Get(ctx, serviceName, clientv3.WithPrefix())
	if err != nil {
		return nil, err
	}
	instances := make([]register.ServiceInstance, len(resp.Kvs))
	for i, kv := range resp.Kvs {
		var si register.ServiceInstance
		err = json.Unmarshal(kv.Value, &si)
		if err != nil {
			log.Println(err)
			continue
		}
		instances[i] = si
	}
	return instances, nil
}

func (r *Register) Subscribe(serviceName string) <-chan register.Event {
	ctx, cancel := context.WithCancel(context.Background())
	r.mu.Lock()
	r.cancels = append(r.cancels, cancel)
	r.mu.Unlock()
	ch := r.client.Watch(ctx, serviceName, clientv3.WithPrefix())
	event := make(chan register.Event)

	go func() {
		for {
			select {
			case <-ctx.Done():
				log.Println(ctx.Err())
				return
			case res := <-ch:
				if res.Err() != nil {
					log.Println(res.Err())
					continue
				}
				if res.Canceled {
					log.Println("etcd: subscribe event canceled")
					continue
				}
				for range res.Events {
					event <- register.Event{}

				}
			}
		}
	}()
	return event
}

func (r *Register) Close() error {
	r.mu.Lock()
	cancels := r.cancels
	cancels = make([]context.CancelFunc, 0)
	r.mu.Unlock()
	for _, cancel := range cancels {
		cancel()
	}
	return r.sess.Close()
}

func NewRegister(client *clientv3.Client) (*Register, error) {
	sess, err := concurrency.NewSession(client)
	if err != nil {
		return nil, err
	}
	return &Register{
		client:  client,
		sess:    sess,
		cancels: make([]context.CancelFunc, 0),
	}, nil
}

func getInstanceName(serviceName string, addr string) string {
	return fmt.Sprintf("/micro/%s/%s", serviceName, addr)
}

func getServiceName(serviceName string) string {
	return fmt.Sprintf("/micro/%s", serviceName)
}

var _ register.Register = (*Register)(nil)
